package com.atguigu.flink.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @author Felix
 * @date 2023/11/29
 * 该案例演示了以流的形式处理指定端口数据(无界)
 */
public class Flink03_stream_unbound {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.从指定的网络端口读取数据
        /*DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.对读取的数据进行转换 封装为一个Tuple2
        SingleOutputStreamOperator<Tuple2<String, Long>> tupleDS = socketDS.flatMap(
            new FlatMapFunction<String, Tuple2<String, Long>>() {
                @Override
                public void flatMap(String lineStr, Collector<Tuple2<String, Long>> out) throws Exception {
                    String[] wordArr = lineStr.split(" ");
                    for (String word : wordArr) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                }
            }
        );
        //TODO 4.按照单词分组
        KeyedStream<Tuple2<String, Long>, Tuple> keyedDS = tupleDS.keyBy(0);
        //TODO 5.聚合
        SingleOutputStreamOperator<Tuple2<String, Long>> sumDS = keyedDS.sum(1);
        //TODO 6.打印
        sumDS.print();*/

        //简化代码
        env
            .socketTextStream("hadoop102", 8888)
            .flatMap(
                (String lineStr, Collector<Tuple2<String, Long>> out) ->{
                    String[] wordArr = lineStr.split(" ");
                    for (String word : wordArr) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                }
            // ).returns(new TypeHint<Tuple2<String, Long>>() {})
            ).returns(Types.TUPLE(Types.STRING,Types.LONG))
            .keyBy(wordTuple->wordTuple.f0)
            .sum(1)
            .print();

        //TODO 7.提交
        env.execute();

    }
}
